package com.heima.article.stream.handler;

import com.alibaba.fastjson.JSON;
import com.heima.article.stream.KafkaStreamListener;
import com.heima.common.constants.message.HotArticleConstants;
import com.heima.model.mess.app.ArticleVisitStreamMess;
import com.heima.model.mess.app.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.springframework.stereotype.Component;

import java.time.Duration;

@Component
@Slf4j
public class HotArticleStreamHandler implements KafkaStreamListener<KStream<String, String>> {
    /**
     * 监听主题
     *
     * @return
     */
    @Override
    public String listenerTopic() {
        return HotArticleConstants.HOTARTICLE_SCORE_INPUT_TOPIC;
    }

    /**
     * 转发到某个主题
     *
     * @return
     */
    @Override
    public String sendTopic() {
        return HotArticleConstants.HOTARTICLE_INCR_HANDLE_OUPUT_TOPIC;
    }

    /**
     * 计算方法执行流程
     *
     * @param stream
     * @return
     */
    @Override
    public KStream<String, String> getService(KStream<String, String> stream) {
        //获取传过来的流
        return stream.map((key, value) -> {
            //创建封装对象
            ArticleVisitStreamMess articleVisitStreamMess = new ArticleVisitStreamMess();
            log.info("开始进行计算，获取行为数据================");
            //获取用户行为
            UpdateArticleMess updateArticleMess = JSON.parseObject(value, UpdateArticleMess.class);
            return new KeyValue<String, String>(updateArticleMess.getArticleId() + ":" + updateArticleMess.getType(), key);
        })      //根据id分组
                .groupByKey()
                //设置聚合时间
                .windowedBy(TimeWindows.of(Duration.ofMinutes(10)))
                //统计分组下总数
                .count()
                //转换成kstream
                .toStream()
                //将处理后的key和value转换成字符串
                .map((key, value) -> {
                    log.info("获取文章id及行为对象 {} ，行为次数为 {}", key.key(), getScoreValue(key.key(), value));
                    return new KeyValue<String, String>(key.key(), getScoreValue(key.key(), value));
                });
    }

    /**
     * 获取用户行为的分值
     *
     * @param key
     * @param value
     * @return
     */
    private String getScoreValue(String key, Long value) {
        //解析key获取文章id和行为
        String[] split = key.split(":");
        String action = split[1];
        //判断行为是哪一种
        ArticleVisitStreamMess articleVisitStreamMess = new ArticleVisitStreamMess();
        articleVisitStreamMess.setArticleId(Long.valueOf(split[0]));
        if (action.equals(UpdateArticleMess.UpdateArticleType.VIEWS.name())) {
            articleVisitStreamMess.setView(value);
        }
        if (action.equals(UpdateArticleMess.UpdateArticleType.LIKES.name())) {
            articleVisitStreamMess.setLike(value);
        }
        if (action.equals(UpdateArticleMess.UpdateArticleType.COMMENT.name())) {
            articleVisitStreamMess.setComment(value);
        }
        if (action.equals(UpdateArticleMess.UpdateArticleType.COLLECTION.name())) {
            articleVisitStreamMess.setCollect(value);
        }
        return JSON.toJSONString(articleVisitStreamMess);
    }
}
